package ltq.stormOld;


import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import ltq.util.*;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;

public class TelTimeCountBolt extends BaseRichBolt {
    private Map<String, Long> telTimeList;
    private int item,itemB;
    Connection conn=null;
    PreparedStatement ps = null;
    ResultSet rs = null;
    String[] nums;

    @Override
    public void prepare(Map<String, Object> map, TopologyContext topologyContext, OutputCollector outputCollector) {
        telTimeList = new HashMap<>();
        item=0;
        itemB=0;
        conn= DatabaseUtil.getConnection();
        nums=new String[100];
    }

    @Override
    public void execute(Tuple tuple) {
        String num_time=tuple.getStringByField("num_time");
        String num=num_time.split(" ")[0];
        if (num.equals("null")){
            System.out.println("-----null------");
            for (String key : telTimeList.keySet()) {
                refreshData(key,telTimeList.get(key));
            }
            System.out.println(item);
            return;
        }
        Long time=Long.parseLong(num_time.split(" ")[1]);

        if (telTimeList.containsKey(num)){
            Long t=telTimeList.get(num);
            time+=t;
        }
        telTimeList.put(num,time);
        item+=1;
        System.out.println(item);

        itemB=Integer.parseInt(num_time.split(" ")[2]);
        if (item%itemB==0){
            for (String key : telTimeList.keySet()) {
                refreshData(key,telTimeList.get(key));
            }
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }

    public void refreshData(String num,Long time){
//        Long time2=0L;
        try {
            ps= conn.prepareStatement("select * from telTest3 where telNum=?");
            ps.setObject(1,num);
            rs= ps.executeQuery();
            if (rs.next()){
//                time2=rs.getLong("totalTime");
                ps= conn.prepareStatement("update telTest3 set totalTime=? where telNum=?");
                ps.setObject(1,time);
//                System.out.println("-----num------"+num);
                ps.setObject(2,num);
                ps.executeUpdate();
            }
            else {
                ps= conn.prepareStatement("insert into telTest3(telNum,totalTime) values (?,?)");
//                System.out.println("-----num------"+num);
                ps.setObject(1,num);
                ps.setObject(2,time);
                ps.executeUpdate();
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }
}
